package study.rocketmq.origin;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class Consumer {
    static Logger logger = LoggerFactory.getLogger(Consumer.class);

    public static final String NAMESRC_ADDR = "120.25.73.46:9876";
    public static void main(String[] args) throws Exception {
        /**
         * TODO
         */
        //参数为 消费者组。
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("Consumer-Group-01");

        defaultMQPushConsumer.setNamesrvAddr(NAMESRC_ADDR);
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 支持多个使用||分割。如 TagA||TagB
        // 如果send中有多个Tag，比如TagA，TagB，消费者只监听了一个TagA，那么TagB也可能被拉取到消费者中，但是会忽略TagB的消息（等于消息丢失了）。
        // 可以通过不同消费者组解决丢失问题。
        // 如果send比如TagA，TagB，消费者只监听了一个TagA，然后同一消费者组的启动TagB，那么TagA无效，TagB有效
        // 原因：同一个消费者组中，后订阅的Tag会覆盖先订阅的Tag。
        defaultMQPushConsumer.subscribe("topic-17","TagA");

        /****
         *  支持SQL方式过滤
         *  数字比较: > < >= <=  = BETWEEN
         *  字符串比较：= IN <>
         *  IS NULL OR IS NOT NULL
         *  逻辑表达式：AND OR NOT
         *  布尔型：TRUE  FALSE
         *  字符串：'' 单引号
         *  NULL
         */
        //defaultMQPushConsumer.subscribe("topic-08", MessageSelector.bySql(" a between 0 and 3"));

        /**TODO 一次性拉取多少消息, 默认32  */
        defaultMQPushConsumer.setPullBatchSize(6);
        /**TODO 一次性拉取PullBatchSize后，分批处理每批的个数 - consumeMessage中List<MessageExt>的个数, 默认1
         *  注意：批次可以并发处理，并发线程见ConsumeThreadMax和setConsumeThreadMin设置
         **/
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(4);
        /**TODO 最大线程数和最小线程数  默认20  */
        defaultMQPushConsumer.setConsumeThreadMax(20);
        defaultMQPushConsumer.setConsumeThreadMin(20);
        /**最大重试次数： 默认会有16次，每次间隔时间：10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h*/
        defaultMQPushConsumer.setMaxReconsumeTimes(1);
        // 集群模式还是广播模式
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                for(MessageExt me : msgs) {
                    try {
                        String topic = me.getTopic();
                        String tags = me.getTags();
                        String keys = me.getKeys();
                        int QueueId = context.getMessageQueue().getQueueId();
                        logger.info(keys + "-开始消费 -- > msgs.size =" + msgs.size());
                        if (keys.equals("key-10")) {
                            int a = 1 / 0;
                        }
                        String msgBody = new String(me.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        logger.info(QueueId + "-消费成功- topic: " + topic + ",tags: " + tags + ", keys: " + keys + ",body: " + msgBody);
                    } catch (Exception e) {
                        e.printStackTrace();
                        /** TODO 重试的次数（不包括第一次失败） */
                        int recousumeTimes = me.getReconsumeTimes();
                        /** TODO 重发3次就不重试了 */
                        if (recousumeTimes == 3) {
                            //		记录日志....
                            //  	做补偿处理
                            /** TODO 设置返回成功，就不重试了。 */
                            logger.error("消费失败-已经重试了【" + recousumeTimes + "】次，不再重试了 -- >");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        } else {
                            logger.error("消费失败-准备重试- 已重试次数: " + recousumeTimes);
                        }
                        /** TODO RabbitMQ会自动重试，共1次，重试时间间隔如下：
                         *     10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                         *     ps：延迟消息多了两档：1s 5s，其余一样。
                         *         messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                         * */
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;


            }
        });
        defaultMQPushConsumer.start();
        System.err.println("consumer start...");
    }
}
